package com.raymond.cmpp.client;

import com.raymond.cmpp.client.msg.MessageManages;
import com.raymond.cmpp.client.netty.ClientHandlersInitializer;
import com.raymond.cmpp.client.netty.ExponentialBackOffRetry;
import com.raymond.cmpp.client.netty.RetryPolicy;
import com.raymond.cmpp.client.queue.PackBlockingQueue;
import com.raymond.cmpp.client.service.Callback;
import com.raymond.cmpp.client.thread.AsyncTaskHandleThread;
import com.raymond.cmpp.client.thread.PackRequestThread;
import com.raymond.cmpp.common.codec.BaseCodecStrategy;
import com.raymond.cmpp.common.msg.BaseMessage;
import com.raymond.cmpp.common.msg.request.TerminateRequestMessage;
import com.raymond.cmpp.common.thread.BaseCirculationThread;
import com.raymond.cmpp.common.utils.SequenceIdUtil;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.util.concurrent.Future;
import io.netty.util.concurrent.GenericFutureListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.ConnectException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;

/**
 * 队列客户端
 *
 * @author :  raymond
 * @version :  V1.0
 * @date :  2021-08-05 11:39
 */
public class CmppClient {
    private final static Logger logger = LoggerFactory.getLogger(CmppClient.class);
    /** ip **/
    private String host;
    /** port **/
    private int port;
    /** 响应超时时间 **/
    private int timeout;
    /**
     * 是否主动关闭连接
     */
    private boolean close;
    /**
     * 源地址，此处为SP_Id，即SP的企业代码。
     */
    private String sourceAddr;

    private String password;

    private final Bootstrap bootstrap = new Bootstrap();
    /**
     * 消息管理
     */
    final ConcurrentHashMap<Integer, MessageManages> msgManager = new ConcurrentHashMap<>();

    final BlockingQueue<MessageManages> asyncTask = new LinkedBlockingQueue<>();

    private int version = 0X30;

    private boolean isConnect = false;
    /** -1:连接失败,
     * -2:登录无响应,
     * 0:登录成功
     * 1：消息结构错
     * 2：非法源地址
     * 3：认证错
     * 4：版本太高
     * 5~ ：其他错误**/
    private int loginStatus = -1;

    /** 重连策略 */
    private RetryPolicy retryPolicy;
    /** 将<code>Channel</code>保存起来, 可用于在其他非handler的地方发送数据 */
    Channel channel;

    final List<BaseCirculationThread> threads = new ArrayList<>();

    public synchronized void close() {
        if (this.close) {
            return;
        }
        this.close = true;
        this.isConnect = false;
        syncSend(new TerminateRequestMessage(SequenceIdUtil.getSequenceId()));
        threads.forEach(BaseCirculationThread::currentThreadExit);
    }

    public boolean isClose() {
        return close;
    }

    public CmppClient(String host, int port, String sourceAddr, String password) {
        this(host, port, sourceAddr, password, 0X30);
    }

    public CmppClient(String host, int port, String sourceAddr, String password, int version) {
        this(host, port, sourceAddr, password, version, 10000);
    }
    public CmppClient(String host, int port, String sourceAddr, String password, int version, int timeout) {
        this(host, port, sourceAddr, password, version, timeout, new ExponentialBackOffRetry(1000, Integer.MAX_VALUE, 60 * 1000));
    }
    public CmppClient(String host, int port, String sourceAddr, String password, int version, int timeout, RetryPolicy retryPolicy) {
        this.host = host;
        this.port = port;
        this.password = password;
        this.timeout = timeout;
        this.retryPolicy = retryPolicy;
        this.sourceAddr = sourceAddr;
        this.version = version;
        init();
    }


    public String getSourceAddr() {
        return sourceAddr;
    }

    public void setSourceAddr(String sourceAddr) {
        this.sourceAddr = sourceAddr;
    }

    public String getPassword() {
        return password;
    }

    public void setPassword(String password) {
        this.password = password;
    }

    public int getVersion() {
        return version;
    }

    public void setVersion(int version) {
        this.version = version;
    }

    public int getTimeout() {
        return timeout;
    }

    public int getLoginStatus() {
        return loginStatus;
    }

    public void setLoginStatus(int loginStatus) {
        this.loginStatus = loginStatus;
    }

    /**
     * 向远程TCP服务器请求连接
     */
    public void connect() {
        synchronized (bootstrap) {
            ChannelFuture future = bootstrap.connect(host, port);
            future.addListener(getConnectionListener());
            this.channel = future.channel();
            channel.config().setWriteBufferWaterMark(new WriteBufferWaterMark(20 * 1024 * 1024, 20 * 1024 * 1024));
        }
    }

    public RetryPolicy getRetryPolicy() {
        return retryPolicy;
    }

    private void init() {
        logger.debug("cmppClient初始化中");
        try {
            BaseCodecStrategy.init("com.raymond.cmpp.client.codec");
            int version3 = 0X30;
            if (version != version3) {
                BaseCodecStrategy.init("com.raymond.cmpp.client.cmpp2");
            }

        } catch (IllegalAccessException | InstantiationException e) {
            throw new RuntimeException("初始化编解码类异常", e);
        }
        asyncThreadInit();
        EventLoopGroup group = new NioEventLoopGroup();
        // bootstrap 可重用, 只需在TcpClient实例化的时候初始化即可.
        bootstrap.group(group)
                .channel(NioSocketChannel.class)
                .handler(new ClientHandlersInitializer(CmppClient.this));
    }

    private void asyncThreadInit() {
        clearAsyncTaskThreadInit();
    }

    private void clearAsyncTaskThreadInit() {
        BaseCirculationThread clearAsyncTask = new AsyncTaskHandleThread(asyncTask,  this);
        threads.add(clearAsyncTask);
        clearAsyncTask.start();
    }


    private ChannelFutureListener getConnectionListener() {
        return future -> {
            if (!future.isSuccess()) {
                future.channel().pipeline().fireChannelInactive();
            }
        };
    }

    public boolean isDone() {
        return isConnect;
    }

    public void connectSuc() {
        logger.info("客户端连接成功");
        this.isConnect = true;
    }

    public synchronized void sync() throws InterruptedException {
        if (isDone()) {
            return;
        }
        wait();
        if (!isDone()) {
            throw new RuntimeException(getConnectMsg());
        }
    }

    public synchronized void sync(long timeout) throws InterruptedException {
        if (isDone()) {
            return;
        }
        wait(timeout);
        if (!isDone()) {
            throw new RuntimeException(getConnectMsg());
        }
    }

    private String getConnectMsg() {
        return loginStatus == -1 ?
                "连接失败,请检查ip、端口,ip:" + host + "、端口:" + port :
                loginStatus == -2 ? "用户登录无响应,账号:" + sourceAddr + "密码:" + password :
                        "登录失败，请检查账号密码,账号:" + sourceAddr + "密码:" + password + ",登录状态:" + +loginStatus;
    }

    public synchronized void notifyAll0() {
        notifyAll();
    }
    public boolean isActive() {
        return channel.isActive();
    }

    public ChannelFuture writeAndFlush(BaseMessage baseMessage) {
        connectCheck();
        return channel.writeAndFlush(baseMessage);
    }


    private void connectCheck() {
        if (!isActive()) {
            throw new RuntimeException(new ConnectException("连接已断开"));
        }
    }

    public MessageManages syncSend(BaseMessage req) {
        connectCheck();
        MessageManages messageManages = new MessageManages(req);
        msgManager.put(req.getSequenceId(), messageManages);
        writeAndFlush(req).addListener(future -> {
            if (!future.isSuccess()) {
                messageManages.setCode(-1);
                messageManages.notify0();
            }
        });
        messageManages.wait0(timeout);
        msgManager.remove(req.getSequenceId());
        return messageManages;
    }
    @SuppressWarnings("unchecked")
    public MessageManages syncSend(MessageManages messageManages) {
        return syncSendListener(messageManages);
    }
    @SuppressWarnings("unchecked")
    public MessageManages syncSendListener(MessageManages messageManages, GenericFutureListener<? extends Future<? super Void>>... listeners) {
        connectCheck();
        msgManager.put(messageManages.getReq().getSequenceId(), messageManages);
        writeAndFlush(messageManages.getReq()).addListeners(listeners);
        messageManages.wait0(timeout);
        msgManager.remove(messageManages.getReq().getSequenceId());
        return messageManages;
    }

    public void asyncSend(BaseMessage req, Callback callback) {
        asyncSend(new MessageManages(req, callback, timeout));
    }
    @SuppressWarnings("unchecked")
    public void asyncSend(MessageManages messageManages) {
        asyncSendListener(messageManages);
    }

    @SuppressWarnings("unchecked")
    public void asyncSendListener(MessageManages messageManages, GenericFutureListener<? extends Future<? super Void>>... listeners) {
        if (!channel.isWritable()) {
            System.out.println("缓冲区满了");
        }
        msgManager.put(messageManages.getReq().getSequenceId(), messageManages);
        writeAndFlush(messageManages.getReq()).addListener(future -> {
            if (future.isSuccess()) {
                asyncTask.add(messageManages);
            } else if (messageManages.isRetry()){
                asyncSend(messageManages);
            } else {
                msgManager.remove(messageManages.getReq().getSequenceId());
                messageManages.setCode(-1);
                messageManages.asyncSendFail();
            }
        });
    }


    public ConcurrentHashMap<Integer, MessageManages> getMsgManager() {
        return this.msgManager;
    }


}
